package edu.gatech.ece6102;

import edu.gatech.ece6102.beans.AddressBean;
import edu.gatech.ece6102.beans.Message;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.LinkedList;
import java.util.Queue;

/**
 * User: deepal
 * Date: Sep 28, 2010
 * Time: 5:22:47 PM
 */
public class LoadBalancer implements Runnable {

    private static Log LOG = LogFactory.getLog(LoadBalancer.class);
    private ServerSocket serversocket = null;
    private volatile boolean destroyed = false;
    private Queue<Message> messageQueue;
    private LogbasedExecutor logbasedExecutor;
    private Logger messageLogger ;

    public static void main(String[] args) {
        if(args.length < 6) {
            System.out.println("<PORT> <LOGGER_IP> <LOGGER_PORT> <FM_IP> <FM_PORT> <APP_SERVER_PORT>");
            System.exit(0);
        }
        int port = Integer.parseInt(args[0]);
        String loggerIp = args[1];
        int loggerPort = Integer.parseInt(args[2]);
        String fmIp = args[3];
        int fmPort = Integer.parseInt(args[4]);
        int appPort = Integer.parseInt(args[5]);
        LoadBalancer lb = new LoadBalancer(port, loggerIp, loggerPort, fmIp, fmPort, appPort);
        lb.run();
    }

    final int port;
    public LoadBalancer(int port, String loggerIp, int loggerPort, String fmIP, int fmPort, int appServerPort) {
        this.port = port;
        //Message queue
        messageQueue = new LinkedList<Message>();
        try {
            messageLogger = new NetworkLogger(loggerIp, loggerPort);
        } catch (Exception e) {
            System.err.println("Cloud not connect to Logger");
            messageLogger = new MokLogger();
        }
        logbasedExecutor = new LogbasedExecutor(messageQueue, messageLogger);
        //To connect to the fault manager
        new UDPClient(fmIP, fmPort, appServerPort, logbasedExecutor);
    }

    public void run() {
        Thread tt = new Thread(logbasedExecutor);
        tt.start();
        System.out.println("About to run");
        try {
            while (!Thread.interrupted()) {
                try {
                    if (serversocket == null || serversocket.isClosed()) {
                        serversocket = new ServerSocket(port);
                    }
                    //Waiting for a connection from clients
                    Socket socket = this.serversocket.accept();

                    //Build the message from the input stream
                    byte[] messageString = readAllFromInputSteam(socket.getInputStream());
                    //Store the message in the logger
                    String id = messageLogger.store(messageString);
                    // Create the Message object with the byte array and the id
                    Message message = new Message(socket, messageString, id);
                    // add the message to the queue
                    messageQueue.add(message);
                    synchronized (logbasedExecutor) {
                        // notifying the logbasedExecutor
                        logbasedExecutor.notify();
                    }
                    if (LOG.isDebugEnabled()) {
                    }
                } catch (java.io.InterruptedIOException ie) {
                    continue;
                } catch (Throwable ex) {
                    if (Thread.interrupted()) {
                        continue;
                    }
                }
            }
        } finally {
            destroy();
        }
    }

    public void close() throws IOException {
        if (this.serversocket != null) {
            this.serversocket.close();
        }
    }

    public void destroy() {
        this.destroyed = true;
        try {
            close();
        } catch (IOException ex) {
            if (LOG.isWarnEnabled()) {
                LOG.warn("I/O error closing listener", ex);
            }
        }
    }


    /**
     * Create a byte array from an input stream
     * @param in
     * @return
     * @throws IOException
     */
    private byte[] readAllFromInputSteam(InputStream in) throws IOException {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        byte buf[] = new byte[4096];
        try {
            int n = in.read(buf);
            out.write(buf, 0, n);
            out.flush();

        } catch (IOException e) {
        } finally {
            try {
                out.close();
            } catch (IOException e) {
            }
        }
        return out.toByteArray();
    }


}
